package com.spzx.common.rabbitmq.config;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbitmq.domain.SpzxCorrelationData;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;

import java.util.concurrent.TimeUnit;

@Configuration
@Slf4j
public class SpzxMqConfig {
    //配置rabbitmq生产者确认
//生产者可以使用模板类发送mq消息
    @Resource
    RabbitTemplate rabbitTemplate;
    @Resource
    RedisTemplate redisTemplate;
    /*
        rabbitmq消息可靠性： 为什么不可靠(生产者发送消息 到rabbitmq服务端  消费者从mq服务端获取消息使用 然后mq删除被消费过的消息)
            1、生产者发送消息可靠性：
                生产者确认(生产者发送消息时 需要mq服务端ack  生产者判断消息是否成功到交换机 或队列..)
                -  配置：
                    配置文件中开启生产者ack
                    给rabbitTemplate设置生产者确认回调
            2、rabbitmq服务端保存消息的可靠性：
                mq默认在内存中保存交换机 队列 消息
                mq重启宕机都会导致内存中的数据丢失
                启用交换机队列消息持久化
            3、消费者消费消息可靠性
                消费者默认接收到消息时给rabbitmq确认 mq服务端删除该消息
                    - 如果消费者消费异常/宕机 都会导致消息未能正常使用 导致消息丢失
                消费者启用手动ack，在代码中正常使用消息以后 手动发送ack，此时mq才会删除该消息

     */
    @PostConstruct //java原生注解  标注的方法可以在当前类构造器调用后 执行
    public void init(){
        //1、消息到达交换机的确认回调
        rabbitTemplate.setConfirmCallback((@Nullable CorrelationData correlationData,
                                           boolean ack, @Nullable String cause)->{
            //correlationData：表示发送消息时配置的消息对象，可以用来携带消息内容 便于重试发送消息
            //  默认是空，需要我们发送消息时手动设置。
            if(!ack){
                //可以尝试重试发送消息
                retrySendMsg(correlationData.getId());
            }
        });
        //2、消息未到达队列的回调
        rabbitTemplate.setReturnsCallback((ReturnedMessage returned)->{
            //可以尝试重试发送消息
            //1、获取redis中的重试消息对象: ReturnedMessage的MessageProperties的头中会携带 spzxCorrelationData的id值
            MessageProperties properties = returned.getMessage().getMessageProperties();
            String msgId = properties.getHeader("spring_returned_message_correlation");
            retrySendMsg(msgId);

        });
    }


    /*
        延迟关单的 业务交换机队列绑定 死信队列交换机绑定
     */
    //死信交换机队列绑定
    @Bean
    public Exchange spzxCloseOrderDeadExchange(){
        return ExchangeBuilder.topicExchange("spzx.close.order.dead.exchange")
                .durable(true)
                .build();
    }
    @Bean
    public Queue spzxCloseOrderDeadQueue(){
        return QueueBuilder
                .durable("spzx.close.order.dead.queue")
                .build();
    }
    //{"spzx.close.order.dead.queue"}
    @Bean
    public Binding spzxCloseOrderDeadBinding(Exchange spzxCloseOrderDeadExchange,
                                             Queue spzxCloseOrderDeadQueue){
        return BindingBuilder
                .bind(spzxCloseOrderDeadQueue)
                .to(spzxCloseOrderDeadExchange)
                //死信路由key
                .with("spzx.close.order.dead.msg")
                .noargs();
    }


    //业务交换机队列绑定
    @Bean
    public Exchange spzxCloseOrderExchange(){
        return ExchangeBuilder.topicExchange("spzx.close.order.exchange")
                .durable(true)
                .build();
    }
    //"spzx.close.order.dead.msg"
    @Bean
    public Queue spzxCloseOrderQueue(){//需要设置过期时间 以及绑定死信队列
        return QueueBuilder
                .durable("spzx.close.order.queue")
                .ttl(30*60*1000)//30*1000)//30*60*1000) //过期时间为30分钟
                //绑定死信交换机
                .deadLetterExchange("spzx.close.order.dead.exchange")
                //死信路由key：
                .deadLetterRoutingKey("spzx.close.order.dead.msg")
                .build();
    }
    @Bean
    public Binding spzxCloseOrderBinding(Exchange spzxCloseOrderExchange,
                                             Queue spzxCloseOrderQueue){
        return BindingBuilder
                .bind(spzxCloseOrderQueue)
                .to(spzxCloseOrderExchange)
                .with("spzx.close.order.msg")
                .noargs();
    }
    //以后 锁定库存成功 发送消息到交换机spzxCloseOrderExchange,消息路由key为spzx.close.order.msg
    // 消息会到达队列：spzxCloseOrderQueue
    //  spzxCloseOrderQueue队列中的消息如果未被消费，30分钟后会自动过期
    // 过期的消息会通过spzx.close.order.dead.exchange 使用路由key spzx.close.order.dead.msg 发送到死信队列中
    // spzxCloseOrderDeadQueue接收到消息
    // 如果有消费者监听spzxCloseOrderDeadQueue 就可以再30分钟后获取消息使用



    private void retrySendMsg(String msgId) {
        SpzxCorrelationData spzxCorrelationData = (SpzxCorrelationData) redisTemplate.opsForValue()
                .get("spzx:mq:retry:" + msgId);
        //需要重试
        if(spzxCorrelationData.isRetry()){
            spzxCorrelationData.setRetryCount(spzxCorrelationData.getRetryCount()+1);
            //将修改了重试次数的对象同步到redis中
            redisTemplate.opsForValue().set("spzx:mq:retry:"+spzxCorrelationData.getId(),
                    spzxCorrelationData,600,TimeUnit.SECONDS);
            if(spzxCorrelationData.getRetryCount()> spzxCorrelationData.getRetryLimit()){
                //删除重试缓存
                redisTemplate.delete("spzx:mq:retry:"+spzxCorrelationData.getId());
                //重试次数超过允许次数:停止重试发送消息
                log.error("消息发送失败，重试次数超过限制，消息内容:{}", JSON.toJSONString(spzxCorrelationData));
            }else{
                //重试发送消息
                rabbitTemplate.convertAndSend(spzxCorrelationData.getExchange(),
                        spzxCorrelationData.getRoutingKey(),
                        spzxCorrelationData.getData() , spzxCorrelationData);
                log.warn("消息发送失败,重试发送消息,消息内容:{}",JSON.toJSONString(spzxCorrelationData));
            }
        }else{
            //删除redis中缓存的消息内容
            redisTemplate.delete("spzx:mq:retry:"+spzxCorrelationData.getId());
            log.error("消息发送失败，不尝试重试，消息内容:{}", JSON.toJSONString(spzxCorrelationData));
        }
    }
}
